package main

import (
    "bufio"
    "context"
    "encoding/base64"
    "errors"
    "fcs-api/src/app/business"
    "fcs-api/src/app/handler"
    "fcs-api/src/app/model"
    "fcs-api/src/app/ops"
    "fcs-api/src/app/service/ksvc"
    "fcs-api/src/app/utils"
    "fcs-api/src/app/utils/resource"
    "fcs-api/src/client"
    "fcs-api/src/client/cks/client"
    cksresource "fcs-api/src/client/cks/resource"
    "fcs-api/src/client/tms"
    "fcs-api/src/monitor/data"
    "fmt"
    "io"
    "log"
    "net"
    "net/http"
    "os"
    "path/filepath"
    "runtime"
    "strconv"
    "strings"
    "sync"
    "time"

    cksClient "fcs-api/src/client/cks/client"

    "git.inspur.com/x/common/rest/tracing"
    "github.com/cenkalti/backoff/v4"
    "k8s.io/api/core/v1"
    apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
    k8serrors "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/util/intstr"
    kube "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"

    "encoding/json"
    "io/ioutil"

    "git.inspur.com/x/auditlog"
    "git.inspur.com/x/clustermanager-client/cluster"
    "git.inspur.com/x/common/loggerfactory"
    "git.inspur.com/x/common/validator"
    "git.inspur.com/x/golang-sql-migrate/migrate"
    adapterClient "git.inspur.com/x/iam-adapter-go/client"
    "git.inspur.com/x/keycloak"
    "github.com/alecthomas/log4go"
    "github.com/emicklei/go-restful"
    "github.com/emicklei/go-restful-openapi"
    "github.com/go-openapi/spec"
    "github.com/spf13/pflag"
    "k8s.io/apimachinery/pkg/util/wait"
    "knative.dev/pkg/signals"

    "git.inspur.com/x/common/i18n"
    "git.inspur.com/x/http-executor-adapter-go"
    "git.inspur.com/x/models"

    "database/sql"
    _ "github.com/go-sql-driver/mysql"
    "github.com/minio/minio-go/v7"
    "github.com/minio/minio-go/v7/pkg/credentials"
    _ "go.uber.org/automaxprocs"
)

const (
    errorPrefix   = "201"
    apiserverPort = 443
)

var (
    argInsecurePort          = pflag.Int32("insecure-port", 9090, "The port to listen to for incoming HTTP requests.")
    argPort                  = pflag.Int("port", 8443, "The secure port to listen to for incoming HTTPS requests.")
    argInsecureBindAddress   = pflag.IP("insecure-bind-address", net.IPv4(127, 0, 0, 1), "The IP address on which to serve the --port (set to 0.0.0.0 for all interfaces).")
    argBindAddress           = pflag.IP("bind-address", net.IPv4(0, 0, 0, 0), "The IP address on which to serve the --secure-port (set to 0.0.0.0 for all interfaces).")
    argDefaultCertDir        = pflag.String("default-cert-dir", "/certs", "Directory path containing '--tls-cert-file' and '--tls-key-file' files. Used also when auto-generating certificates flag is set.")
    argCertFile              = pflag.String("tls-cert-file", "", "File containing the default x509 Certificate for HTTPS.")
    argKeyFile               = pflag.String("tls-key-file", "", "File containing the default x509 private key matching --tls-cert-file.")
    apiServerIps             = pflag.String("apiserver-ips", "100.105.0.1", "api server apis")
    keyCloakConfigFile       = pflag.String("keycloak-config-file", "../config/keycloak.conf", "The keycloak config file.")
    iamApiServerHost         = pflag.String("iam-apiserver-host", "http://iam-apiserver.iam:9101", "auth api server address.")
    iamAkSkHost              = pflag.String("iam-apiserver-host-special", "", "get aksk server address.")
    iamLadonHost             = pflag.String("iam-ladon-host", "http://iam-auth-ladon.iam:8090", "iam london host")
    region                   = pflag.String("region", "cn-north-3", "region.")
    clusterName              = pflag.String("cluster", "", "")
    knativeClusterUser       = pflag.String("knative-cluster-user", "pro", "knative-cluster-user.")
    knativeClusterP6d        = pflag.String("knative-cluster-passwd", "pro", "knative-cluster-passwd.")
    dbSettingFile            = pflag.String("db-setting", "", "The database setting file.")
    xormLogRootPath          = pflag.String("xorm-log-root", "", "The xorm log root path.")
    imageRepository          = pflag.String("image-repository", "10.10.4.181:5000", "image repository")
    imageRepositorySpecial   = pflag.String("image-repository-special", "", "image repository")
    cirEndpoint              = pflag.String("cir-endpoint", "", "cir endpoint")
    runtimeImage             = pflag.String("runtime-image", "", "runtime image tag")
    functionInvokeHost       = pflag.String("function-invoke-host", "", "function invoke host url")
    httpInvokeHost           = pflag.String("http-invoke-host", "", "http invoke host")
    gatewayNamespace         = pflag.String("gateway-namespace", "knative-kong", "gateway namespace")
    downloadHost             = pflag.String("oss-download-host", "https://cnp-dev.inspurcloud.cn/", "oss-download-host")
    ossAdminHost             = pflag.String("oss-apiserver-host", "mysql-minio.fcs.svc.cluster.local:9000", "oss-admin-host")
    ossUserHost              = pflag.String("oss-apiserver-host-special", "", "oss-user-host")
    ossEndpoint              = pflag.String("oss-inner-endpoint-special", "", "oss-endpoint")
    ossInnerEndpoint         = pflag.String("oss-inner-endpoint", "mysql-minio.fcs.svc.cluster.local:9000", "oss intranet endpoint")
    ossBucket                = pflag.String("oss-bucket", "fcs-code", "oss-bucket")
    minioAk                  = pflag.String("minio-ak", "minioadmin", "minioAk")
    minioSk                  = pflag.String("minio-sk", "minioadmin", "minioSk")
    apigHost                 = pflag.String("apig-apiserver-host", "http://apig-api.cnp-system:8080/", "apig-host")
    jobCenterHost            = pflag.String("job-center-host", "http://job-center.common.svc.cluster.local:8080", "job center host")
    argCrosAllowedDomain     = pflag.String("cors-allowed-domains", "*", "list of allowed values for Http Origin. If empty all are allowed.")
    ksvcWaitingTime          = pflag.String("ksvc-waiting-time", "120", "the maximum time waiting ksvc ready.")
    esHost                   = pflag.String("es-host", "http://elasticsearch.logging.svc.cluster.local:9200", "elasticsearch service host.")
    logDelayTime             = pflag.String("log-delay-time", "150", "log delay time.")
    callbackUser             = pflag.String("callback-user", "", "callback user info.")
    callbackEndpoint         = pflag.String("callback-endpoint", "", "callback host.")
    functionBlockCidr        = pflag.String("function-block-cidr", "", "function can't access these ip address.")
    migrateServerHost        = pflag.String("migrate-server-host", "http://liquibase-server.cnp-system:80/", "migrate server address.")
    networkpolicyWhitelistIp = pflag.String("networkpolicy-whitelist-ip", "", "the ip list need set in networkpolicy")
    encrypted                = pflag.Bool("audit-encrypted", false, "audit encrypted")
    publicKey                = pflag.String("audit-public-key", "", "publickey")
    vpcHost                  = pflag.String("vpc-apiserver-host", "", "vpc service backend address")
    eipHost                  = pflag.String("eip-apiserver-host", "http://eip-service.network:8080/", "eip service backend address")
    iceHost                  = pflag.String("ice-apiserver-host", "http://cks-core-api.cks:9091/", "ice service backend address")
    ecsHost                  = pflag.String("ecs-apiserver-host", "http://ecs-service.compute:8080/", "ecs service backend address")
    clusterMasterP6d         = pflag.String("cluster-master-passwd", "", "master node password")
    influxdbHost             = pflag.String("influxdb-server-host", "http://influxdb.fcs:9096/", "influxdb url address")
    influxdbUser             = pflag.String("influxdb-server-user", "admin", "influxdb user")
    influxdbP6d              = pflag.String("influxdb-server-passwd", "FA79nfx3UP", "influxdb password")
    freeCount                = pflag.Int64("free-count", 1000000, "config cost free count")
    freeComputePower         = pflag.Float64("free-compute", 400000, "config cost free compute power")
    cloudType                = pflag.String("cloud-type", "native", "cloud-type: native/public")
    cecURL                   = pflag.String("cecURL", "pub1:pub1", "")

    // bss mq相关参数
    consoleMqHost     = pflag.String("console-mq-host", "bss-forward-mq.common.svc.cluster.local", "consoleMqHost")
    consoleMqExchange = pflag.String("console-mq-exchange", "exchange.region", "ConsoleMqUrl")

    consoleMqOrderQueue            = pflag.String("console-mq-order-queue", "queue.fcs.order", "ConsoleMqOrderQueue")
    consoleMqOrderRouteKey         = pflag.String("console-mq-order-rk", "key.fcs.order", "ConsoleMqOrderRouteKey")
    consoleMqOrderCallbackQueue    = pflag.String("console-mq-order-callback-queue", "queue.return.order", "ConsoleMqOrderCallbackQueue")
    consoleMqOrderCallbackRouteKey = pflag.String("console-mq-order-callback-rk", "key.return.order", "ConsoleMqOrderCallbackRouteKey")

    consoleMqChangeQueue            = pflag.String("console-mq-change-queue", "queue.fcs.change", "ConsoleMqChangeQueue")
    consoleMqChangeRouteKey         = pflag.String("console-mq-change-rk", "key.fcs.change", "ConsoleMqChangeRk")
    consoleMqChangeCallbackQueue    = pflag.String("console-mq-change-callback-queue", "queue.return.change", "ConsoleMqChangeCallbackQueue")
    consoleMqChangeCallbackRouteKey = pflag.String("console-mq-change-callback-rk", "key.return.change", "ConsoleMqChangeCallbackRk")

    consoleMqBillQueue    = pflag.String("console-mq-bill-queue", "queue.return.bill", "ConsoleMqBillQueue")
    consoleMqBillRouteKey = pflag.String("console-mq-bill-rk", "key.return.bill", "ConsoleMqBillRk")

    consoleMqMessageTTL         = pflag.Int("console-mq-message-ttl", 300000, "ConsoleMqMessageTTL")
    consoleMqDeadLetterExchange = pflag.String("console-mq-dead-letter-exchange", "exchange.dlx", "ConsoleMqDeadLetterExchange")
    consoleMqDeadLetterRouteKey = pflag.String("console-mq-dead-letter-rk", "key.dlx", "ConsoleMqDeadLetterRk")

    defaultLanguage = pflag.String("default-language", "zh", "i18n default language.")
)

var isInitKong = false
var isInitExternalIP = false
var isInitLayerConfig = false

var logger = loggerfactory.GetLogger()

func initArgsFromFile() {
    filePath := "./src/config/fcs-app.conf"
    for _, arg := range os.Args {
        if strings.Index(arg, "fcs-config-file=") > 0 && len(arg) > 22 {
            filePath = strings.Split(arg, "=")[1]
        }
    }
    if filePath == "" {
        return
    }
    fi, err := os.Open(filePath)
    if err != nil {
        log.Fatalf("Error load  config file: %v", err)
        return
    }
    defer func() {
        if err := fi.Close(); err != nil {
            _ = logger.Error("Error closing file: %s\n", err)
        }
    }()
    br := bufio.NewReader(fi)
    args := make([]string, 0)
    for {
        a, _, c := br.ReadLine()
        if c == io.EOF {
            break
        }
        arg := string(a)
        if strings.Index(arg, "--") == 0 && strings.Index(arg, "=") > 0 {
            argName := arg[2:strings.Index(arg, "=")]
            flg := pflag.CommandLine.Lookup(argName)
            if flg != nil {
                args = append(args, arg)
            }
        }
    }
    log.Println("loaded config file")

    err = pflag.CommandLine.Parse(args)
    if err != nil {
        println(err.Error())
        log.Fatalf("Error load fcs config file: %v", err)
    }
    log.Println("parsed config file finish")
}

func initDbConfig(dbSettingFile, xormLogRootPath string) {
    _ = models.InitEngine(dbSettingFile, xormLogRootPath)
}

func main() {
    log.SetOutput(os.Stdout)
    defer flush(logger)
    initArgsFromFile()
    initAuditLog() // 初始化审计配置
    initDbConfig(*dbSettingFile, *xormLogRootPath)
    go initLiquibase()
    initI18n(*defaultLanguage)
    wsContainer := restful.NewContainer()
    k8sCluster := &cluster.K8sInstance{}
    apiServerIpsArr := strings.Split(*apiServerIps, ",")
    runtimeMap := make(map[string]string)
    err := json.Unmarshal([]byte(*runtimeImage), &runtimeMap)
    if err != nil {
        log.Fatalf("DecodeString runtimeTag error: %v", err)
    }

    //mock数据下沉
    go http_executor.ExecutorInit()

    // 系统结束信号
    sigCh := signals.SetupSignalHandler()

    tokenUrl := iamAkSkHost
    if *tokenUrl == "" {
        tokenUrl = iamApiServerHost
    }
    err = cksclient.InitCksService(*tokenUrl)
    if err != nil {
        log.Fatalf("Error while initializing connection to cks apiserver. Reason: %v", err)
    }

    keycloakClient, err := keycloak.NewForFile(*keyCloakConfigFile)
    if err != nil {
        log.Fatalf("Error while initializing connection to keycloak apiserver. "+
            "This most likely means that load keycloak public key error. Reason: %v", err)
    }

    log.Println("Successful init keycloak client")
    validator.Init(errorPrefix)
    err = adapterClient.Init(keycloakClient, errorPrefix, *iamApiServerHost, *iamLadonHost, *region, "", "", "")
    if err != nil {
        log.Fatalf("adapterClient.Init error : %v", err)
    }

    knativeClusterUser, err := base64.StdEncoding.DecodeString(*knativeClusterUser)
    if err != nil {
        log.Fatalf("DecodeString knativeClusterUser error: %v", err)
        return
    }
    knativeClusterP6d, err := base64.StdEncoding.DecodeString(*knativeClusterP6d)
    if err != nil {
        log.Fatalf("DecodeString knativeClusterPasswd error: %v", err)
        return
    }

    var user []byte
    if *callbackUser == "" {
        user = knativeClusterUser
    } else {
        user, err = base64.StdEncoding.DecodeString(*callbackUser)
        if err != nil {
            log.Fatalf("DecodeString callbackUser error: %v", err)
            return
        }
    }

    log.Printf("knativeClusterUser is %s knativeClusterPasswd is %s", knativeClusterUser, knativeClusterP6d)
    log.Printf("callbackUser is %s", user)

    masterP6d, err := base64.StdEncoding.DecodeString(*clusterMasterP6d)
    if err != nil {
        log.Fatalf("DecodeString clusterMasterPasswd error: %v", err)
        return
    }
    waitingTime, err := strconv.Atoi(*ksvcWaitingTime)
    if err != nil {
        log.Fatalf("DecodeString ksvcWaitingTime error: %v", err)
        return
    }
    if waitingTime == 0 {
        waitingTime = model.KsvcReadyTime
    }
    logDelayTime, err := strconv.Atoi(*logDelayTime)
    if err != nil {
        log.Fatalf("DecodeString logDelayTime error: %v", err)
        return
    }

    // 设置 10000 的缓冲
    invokeChan := make(chan *model.Invoke, 100)
    go business.InvokeCallback(invokeChan)

    oe := *ossInnerEndpoint
    //if !strings.Contains(*ossInnerEndpoint, "http") {
    //	oe = "http://" + *ossInnerEndpoint
    //}
    if *ossEndpoint != "" {
        oe = *ossEndpoint
    }
    imageUrl := *imageRepository
    if *imageRepositorySpecial != "" {
        imageUrl = *imageRepositorySpecial
    }
    queue := make(chan model.ListenInfo, 100)
    model.Base = model.BaseInfo{
        KnativeClusterUser:     string(knativeClusterUser),
        KnativeClusterPwd:      string(knativeClusterP6d),
        ContainerImageRegistry: imageUrl,
        CirEndpoint:            *cirEndpoint,
        RuntimeImage: model.RuntimeImage{
            Python36RuntimeUrl:   imageUrl + runtimeMap[model.Python36],
            NodeJS1022RuntimeUrl: imageUrl + runtimeMap[model.NodeJs1022],
            Python27RuntimeUrl:   imageUrl + runtimeMap[model.Python27],
            NodeJS1218RuntimeUrl: imageUrl + runtimeMap[model.NodeJS1218],
            Golang1RuntimeUrl:    imageUrl + runtimeMap[model.Golang1],
            Java8RuntimeUrl:      imageUrl + runtimeMap[model.Java8],
            PhpRuntimeUrl:        imageUrl + runtimeMap[model.PHP],
        },
        BssMqInfo: model.BssMqInfo{
            Url:                    *consoleMqHost,
            Exchange:               *consoleMqExchange,
            OrderQueue:             *consoleMqOrderQueue,
            OrderRouteKey:          *consoleMqOrderRouteKey,
            OrderCallbackQueue:     *consoleMqOrderCallbackQueue,
            OrderCallbackRouteKey:  *consoleMqOrderCallbackRouteKey,
            ChangeQueue:            *consoleMqChangeQueue,
            ChangeRouteKey:         *consoleMqChangeRouteKey,
            ChangeCallbackQueue:    *consoleMqChangeCallbackQueue,
            ChangeCallbackRouteKey: *consoleMqChangeCallbackRouteKey,
            BillQueue:              *consoleMqBillQueue,
            BillRouteKey:           *consoleMqBillRouteKey,
            MessageTTL:             *consoleMqMessageTTL,
            DeadLetterExchange:     *consoleMqDeadLetterExchange,
            DeadLetterRouteKey:     *consoleMqDeadLetterRouteKey,
        },
        FunctionInvokeHost:       *functionInvokeHost,
        IamApiServerHost:         *iamApiServerHost,
        HttpInvokeHost:           *httpInvokeHost,
        GatewayNamespace:         *gatewayNamespace,
        OssDownloadHost:          *downloadHost,
        OssAdminHost:             *ossAdminHost,
        OssUserHost:              *ossUserHost,
        OssEndpoint:              oe,
        OssBucket:                *ossBucket,
        OssAK:                    *minioAk,
        OssSK:                    *minioSk,
        ApigHost:                 *apigHost,
        KsvcWaitingTime:          waitingTime,
        ESHost:                   *esHost,
        LogDelayTime:             logDelayTime,
        CallbackUser:             string(user),
        CallbackEndpoint:         *callbackEndpoint,
        NetworkpolicyWhitelistIp: *networkpolicyWhitelistIp,
        FunctionBlockCidr:        *functionBlockCidr,
        VpcHost:                  *vpcHost,
        EipHost:                  *eipHost,
        IceHost:                  *iceHost,
        EcsHost:                  *ecsHost,
        ClusterMasterPwd:         string(masterP6d),
        KeycloakConfig:           keycloakClient.GetClientConfig(),
        KeycloakClient:           keycloakClient,
        InfluxdbHost:             *influxdbHost,
        InfluxdbUser:             *influxdbUser,
        InfluxdbP6d:              *influxdbP6d,
        FreeCount:                *freeCount,
        FreeComputePower:         *freeComputePower,
        Region:                   *region,
        Cluster:                  *clusterName,
        CecURL:                   *cecURL,
        CloudType:                *cloudType,
        Queue:                    queue,
    }
    //初始化标签管理配置
    tms.Init()
    // 使用 channel 判断函数计算集群是否已经准备好
    clusterReadyChan := make(chan bool, 1)
    // 判断apiserver是否有配置
    if *apiServerIps != "" {
        clusterReadyChan <- true
        resource.NeedMonitoring = false
    }
    clusterClientManager := client.NewClusterClientManager(k8sCluster)
    model.Base.ClusterManager = clusterClientManager
    apiHandler := handler.APIHandler{
        ClusterManager: clusterClientManager,
        InvokeChan:     invokeChan,
    }
    ops.ApiHandler = apiHandler
    // 初始化token，第一次进性阻塞操作，防止后面的操作没有token失败
    updateAdminToken()
    if model.Base.AdminToken == "" {
        // 如果第一次没有获取到，结束程序
        log.Fatalf("Can't fetch admin token")
    }
    initMinio()
    // 循环获取token
    go wait.Until(updateAdminToken, time.Minute*5, sigCh)
    // 初始化 ak、sk
    //	go wait.Until(initAdminAkSk, time.Minute, sigCh)

    //stopMqCh := make(chan int)
    //go mq.ListenOssMq(*consoleMqHost, stopMqCh)
    //go bss.ListenBssMq()

    ws := handler.NewApplicationHandler(&apiHandler)
    wsContainer.Add(ws)
    config := restfulspec.Config{
        WebServices:                   wsContainer.RegisteredWebServices(), // you control what services are visible
        APIPath:                       "/ifcs/apidocs.json",
        PostBuildSwaggerObjectHandler: enrichSwaggerObject}
    wsContainer.Add(restfulspec.NewOpenAPIService(config))
    createBasicFilters(wsContainer, *argCrosAllowedDomain)
    http.Handle("/ifcs/", HeaderProcess(wsContainer))

    utilHandler := new(handler.APIHandler)
    webServer := handler.NewInnerHandler(utilHandler)
    utilContainer := handler.NewWebService(webServer)
    http.Handle("/metrics", utilContainer)
    http.Handle("/prometheus", utilContainer)
    http.Handle("/healthz", utilContainer)
    http.Handle("/report", utilContainer)

    errCh := make(chan error)
    var server *http.Server
    go func() {
        // Listen for http or https
        if *argCertFile != "" && *argKeyFile != "" {
            certFilePath := *argDefaultCertDir + string(os.PathSeparator) + *argCertFile
            keyFilePath := *argDefaultCertDir + string(os.PathSeparator) + *argKeyFile
            log.Printf("Serving securely on HTTPS port: %d", *argPort)
            secureAddr := fmt.Sprintf("%s:%d", *argBindAddress, *argPort)
            // Don't forward ErrServerClosed as that indicates we're already shutting down.
            server = &http.Server{Addr: secureAddr}
            if err := server.ListenAndServeTLS(certFilePath, keyFilePath); err != nil && err != http.ErrServerClosed {
                errCh <- fmt.Errorf("fcs server failed: %w", err)
            }
        } else {
            log.Printf("Serving insecurely on HTTP port: %d", *argInsecurePort)
            addr := fmt.Sprintf("%s:%d", *argInsecureBindAddress, *argInsecurePort)
            server = &http.Server{Addr: addr}
            // 修改ListenAndServe为ListenAndServeTLS以支持HTTPS
            if err := server.ListenAndServeTLS(*argCertFile, *argKeyFile); err != nil && err != http.ErrServerClosed {
                errCh <- fmt.Errorf("fcs server failed: %w", err)
            }
        }
    }()

    go business.ConfigCronJob(*jobCenterHost)

    // 当配置文件中没有函数计算apiserver配置时，循环从数据库中检测配置
    go func() {
        if *apiServerIps == "" {
            for {
                list, err := model.GetClusterList()
                if err != nil {
                    data.K8sApiErr = true
                    logger.Error("get cluster list from db error: ", err)
                    time.Sleep(time.Minute)
                    continue
                }
                // 如果数据库中有数据，写入数据到ready channel，并返回
                if len(list) != 0 {
                    logger.Info("found cluster info")
                    apiServerIpsArr = strings.Split(list[0].ApiserverIps, ",")
                    model.Base.ClusterMasterPwd = list[0].ClusterMasterPasswd
                    clusterReadyChan <- true
                    resource.NeedMonitoring = false
                    return
                }
                data.K8sApiErr = true
                time.Sleep(time.Minute)
            }
        }
    }()
    // 调用 ice 接口监听集群是否创建
    resource.MonitoringCluster()
    // 调用 ecs 接口监听转发虚机
    resource.MonitoringEcs()
    // 添加初始化代码逻辑
    if *cloudType == "native" {
        go func() {
            for {
                err := resource.Upload(*minioAk, *minioSk)
                if err != nil {
                    time.Sleep(time.Minute)
                } else {
                    return
                }
            }
        }()
    }

    var prepareOnce sync.Once
    checkApiserverFunc := func() {
        err := checkApiserverHost(apiServerIpsArr, k8sCluster)
        if err != nil {
            data.K8sApiErr = true
            _ = logger.Error("Error while check connection to fcs cluster apiserver. Reason: %s", err.Error())
        } else {
            // 政务云环境初始化lcm配置
            prepareOnce.Do(func() {
                if *cloudType == "native" {
                    //go prepareLcm(apiServerIpsArr[0], &apiHandler)
                }
                //go initKongProxyService(apiServerIpsArr)
                // apiserver正常之后，启动ksvc监听
                go startListenKsvc(&apiHandler, queue, sigCh)
                // 监听异常的ksvc，并进行处理
                go ops.StartWatchTimeoutFunctions()
            })
            data.K8sApiErr = false
            if !isInitKong {
                // 初始化函数计算配置
                err = initFcsClusterConfig(&apiHandler)
                if err != nil {
                    _ = logger.Error("Error while initializing fcs cluster config. Reason: %v", err)
                } else {
                    isInitKong = true
                }
            }
            if !isInitExternalIP && *cloudType == "native" {
                // 初始化kong svc的externalip
                //err = initServiceConfig(&apiHandler, apiServerIpsArr)
                //if err != nil {
                //	_ = logger.Error("Error while initializing fcs svc config. Reason: %v", err)
                //} else {
                isInitExternalIP = true
                //}
            }
            if !isInitLayerConfig {
                // 初始化 layer controller 配置
                //err = ops.InitLayerConfigMap(&apiHandler)
                //if err != nil {
                //	_ = logger.Error("Error while initializing fcs layer controller config. Reason: %v", err)
                //} else {
                isInitLayerConfig = true
                //}
            }
        }
    }

    go func() {
        //<-clusterReadyChan
        wait.Until(checkApiserverFunc, time.Minute, sigCh)
    }()

    select {
    case err := <-errCh:
        _ = logger.Error("Failed to bring up fcs service, shutting down, %v", err)
        logger.Info("flushed log data")
        flush(logger)
        os.Exit(1)
    case <-sigCh:
        logger.Info("Received TERM signal, attempting to gracefully shutdown servers.")
        if server != nil {
            if err := server.Shutdown(context.Background()); err != nil {
                _ = logger.Error("Failed to shutdown fcs server %v", err)
            }
        }
        logger.Info("flushed log data")
        logger.Info("Shutdown complete, exiting...")
    }
}

func createBasicFilters(container *restful.Container, allowedDomains string) {
    corsAllowedMethods := []string{"POST", "GET", "OPTIONS", "PUT", "DELETE", "PATCH"}
    corsAllowedHeaders := []string{"Authorization", "Accept", "Content-Type", "Access-Control-Allow-Origin", "Access-Control-Allow-Headers", "Access-Control-Allow-Methods", "Attributes"}

    corsAllowedDomains := []string{}
    if allowedDomains != "" && allowedDomains != "*" {
        corsAllowedDomains = strings.Split(allowedDomains, ",")
    }

    // cors
    cors := restful.CrossOriginResourceSharing{
        AllowedMethods: corsAllowedMethods,
        AllowedHeaders: corsAllowedHeaders,
        AllowedDomains: corsAllowedDomains,
        CookiesAllowed: true,
        Container:      container,
    }
    container.Filter(cors.Filter)
    logger.Info("CORS filter added.")
    // options
    container.Filter(container.OPTIONSFilter)
    logger.Info("OPTIONS filter added.")
}

func flush(logger log4go.Logger) {
    logger.Close()
    _ = os.Stdout.Sync()
    _ = os.Stderr.Sync()
    // metrics.FlushExporter()
}

func initAuditLog() {
    auditlog.Init("fcs", *encrypted, *publicKey, *region)
}

func initAdminAkSk() {
    token := "bearer " + utils.GetAdminToken()
    akskHost := *iamApiServerHost
    if *iamAkSkHost != "" {
        akskHost = *iamAkSkHost
    }
    url := akskHost + "/v1/secrets"
    r, err := http.NewRequest("GET", url, nil)
    if err != nil {
        _ = logger.Error("initAdminAkSk NewRequest error %v", err)
        return
    }
    r.Header.Set("Authorization", token)
    resp, err := http.DefaultClient.Do(r)
    if err != nil {
        _ = logger.Error("initAdminAkSk Client.Do error %v", err)
        return
    }
    body, err := ioutil.ReadAll(resp.Body)
    if err != nil {
        _ = logger.Error("initAdminAkSk ReadAll error %v", err)
        return
    }
    if resp.StatusCode != 200 {
        _ = logger.Error("initAdminAkSk error %v : %v ", resp.StatusCode, string(body))
        return
    }

    var s4ts []map[string]interface{}
    if err = json.Unmarshal(body, &s4ts); err != nil {
        _ = logger.Error("initAdminAkSk Unmarshal error %v", err)
        return
    }

    if len(s4ts) == 0 {
        _ = logger.Error("initAdminAkSk error, the user didn't have any aksk")
        return
    }

    handler.Ak = s4ts[0]["secretId"].(string)
    handler.Sk = s4ts[0]["secretKey"].(string)

    if handler.Ak == "" || handler.Sk == "" {
        _ = logger.Error("initAdminAkSk error, response body didn't contain aksk, response is %s", string(body))
    }
}

func initServiceConfig(api *handler.APIHandler, mgtIps []string) error {
    fcClient, err := handler.GetFcClient(api)
    if err != nil {
        _ = logger.Error("initServiceConfig: get fcClient error %v", err)
        return err
    }
    labelSelector := fmt.Sprintf("%s=%s", model.MaterNodeLabelKey, "true")
    nodes, err := fcClient.GetNodes(labelSelector)
    if err != nil {
        _ = logger.Error("initServiceConfig: get master node error %v", err)
        return err
    }
    vpcIps := make([]string, 0)
    for _, node := range nodes.Items {
        for _, address := range node.Status.Addresses {
            if address.Type == model.NodeIntIPStr {
                vpcIps = append(vpcIps, address.Address)
            }
        }
    }
    kongSvc, err := fcClient.GetService("knative-kong", "kong-proxy")
    if err != nil {
        _ = logger.Error("initServiceConfig: get kong svc error %v", err)
        return err
    }
    // 设置 kong-proxy 的exteranlip
    kongSvc.Spec.ExternalIPs = append(mgtIps, vpcIps...)
    _, err = fcClient.UpdateService("knative-kong", kongSvc)
    if err != nil {
        _ = logger.Error("initServiceConfig: update kong svc error %v", err)
        return err
    }
    // 去除istio-ingressgateway的externalip，防止和kong-proxy的ip冲突
    istioIngressSvc, err := fcClient.GetService("istio-system", "istio-ingressgateway")
    if err != nil {
        _ = logger.Error("initServiceConfig: get istio ingress svc error %v", err)
        return err
    }
    istioIngressSvc.Spec.ExternalIPs = []string{}
    _, err = fcClient.UpdateService("istio-system", istioIngressSvc)
    if err != nil {
        _ = logger.Error("initServiceConfig: update istio ingress svc error %v", err)
        return err
    }
    return nil
}

func initFcsClusterConfig(api *handler.APIHandler) error {
    fcClient, err := handler.GetFcClient(api)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: get fcClient error %v", err)
        return err
    }

    // KCP plugin-fcs-credential
    KcpCredential, err := fcClient.GetKongClusterPlugin(model.KCPCredential)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: get KcpCredential error %v", err)
        return err
    }
    oe := *ossInnerEndpoint
    //if !strings.Contains(*ossInnerEndpoint, "http") {
    //	oe = "http://" + *ossInnerEndpoint
    //}
    if *ossEndpoint != "" {
        oe = *ossEndpoint
    }
    originConfig, err := json.Marshal(KcpCredential.Config)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: json.Marshal(KcpCredential.Config) error", err)
        return err
    }
    var originConfigData model.KcpProxyConfig
    err = json.Unmarshal(originConfig, &originConfigData)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: json.Unmarshal(originConfig, &originConfigData) error", err)
        return err
    }
    originConfigData.OssEndpoint = oe
    originConfigData.OssAccessKeyId = *minioAk
    originConfigData.OssAccessKeySecret = *minioSk
    originConfigData.OssBucketName = *ossBucket
    originConfigData.CallbackEndpoint = *callbackEndpoint
    originConfigData.CallbackAccessKeyId = model.Base.KeycloakConfig.ClientId
    originConfigData.CallbackAccessKeySecret = model.Base.KeycloakConfig.ClientSecret
    originConfigByte, err := json.Marshal(originConfigData)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: json.Marshal(originConfigData) error", err)
        return err
    }
    // 需要判断配置的信息都有值
    if originConfigData.OssEndpoint == "" ||
        originConfigData.OssAccessKeyId == "" ||
        originConfigData.OssAccessKeySecret == "" ||
        originConfigData.OssBucketName == "" ||
        originConfigData.CallbackEndpoint == "" {
        err = fmt.Errorf("KcpProxyConfig values cannot be null")
        _ = logger.Error("initFcsClusterConfig: %s, %s", err, originConfigByte)
        return err
    }
    KcpCredential.Config = apiextensionsv1.JSON{Raw: originConfigByte}
    _, err = fcClient.UpdateKongClusterPlugin(KcpCredential)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: update KcpCredential error", err)
        return err
    }
    logger.Info("initFcsClusterConfig: KcpProxyConfig is %s", string(originConfigByte))

    // KCP plugin-fcs-iam
    KcpIam, err := fcClient.GetKongClusterPlugin(model.KCPIam)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: get KcpIam error %v", err)
        return err
    }
    publicKey, err := business.GetIamPublicKey(model.Base.KeycloakConfig)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: get KcpIam publicKey error %v", err)
        return err
    }
    iamHost := model.Base.KeycloakConfig.Host
    iamOriginConfig, err := json.Marshal(KcpIam.Config)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: KcpIam.Config.Marshal error", err)
        return err
    }
    var iamOriginConfigData model.KcpIamConfig
    err = json.Unmarshal(iamOriginConfig, &iamOriginConfigData)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: json.Unmarshal(iamOriginConfig, &iamOriginConfigData) error", err)
        return err
    }
    iamOriginConfigData.IamHost = iamHost
    iamOriginConfigData.PublicKey = publicKey
    iamOriginConfigByte, err := json.Marshal(iamOriginConfigData)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: json.Marshal(iamOriginConfigData) error", err)
        return err
    }
    // 需要判断配置的信息都有值
    if iamOriginConfigData.IamHost == "" || iamOriginConfigData.PublicKey == "" {
        err = fmt.Errorf("KcpIamConfig values cannot be null")
        _ = logger.Error("initFcsClusterConfig: %s, %s", err, iamOriginConfigByte)
        return err
    }
    KcpIam.Config = apiextensionsv1.JSON{Raw: iamOriginConfigByte}
    _, err = fcClient.UpdateKongClusterPlugin(KcpIam)
    if err != nil {
        _ = logger.Error("initFcsClusterConfig: update KcpIam error", err)
        return err
    }
    logger.Info("initFcsClusterConfig: KcpIamConfig is %s", string(iamOriginConfigByte))

    return nil
}

// 检测IP+端口是否可用
func checkApiserverHost(ips []string, ins *cluster.K8sInstance) error {

    for _, ip := range ips {
        timeout := time.Second
        conn, err := net.DialTimeout("tcp", net.JoinHostPort(ip, strconv.Itoa(apiserverPort)), timeout)
        if err != nil {
            _ = logger.Warn("checkApiserverHost Connecting error: %s", err.Error())
            continue
        }
        if conn != nil {
            defer func() {
                _ = conn.Close()
            }()
            ins.ApiServer = "https://" + ip + ":" + strconv.Itoa(apiserverPort)
            if ip != model.Base.ClusterMasterIp {
                model.Base.ClusterMasterIp = ip
            }
            logger.Info("checkApiserverHost: using api server host: %s", ins.ApiServer)
            return nil
        }
    }
    _ = logger.Error("checkApiserverHost: no healthy apiserver")
    return errors.New("no healthy apiserver")
}

func initLiquibase() {

    dbType := models.DbCfg.Type
    if strings.ToLower(dbType) == "postgres" {
        dbType = "POSTGRESQL"
    } else if strings.ToLower(dbType) == "mysql" {
        dbType = "MYSQL"
    }

    db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s)/", models.DbCfg.User, models.DbCfg.Passwd, models.DbCfg.Host))
    if err != nil {
        logger.Error(err)
    }
    defer db.Close()

    // 执行 SQL 查询
    _, err = db.Query(fmt.Sprintf("CREATE database %s", models.DbCfg.Name))
    if err != nil {
        logger.Error("create database err: %v", err)
    }

    dbHost := strings.Split(models.DbCfg.Host, ":")
    dbPort := 3306
    if len(dbHost) >= 2 {
        dbPort, _ = strconv.Atoi(dbHost[1])
    }
    dataSource := migrate.DataSource{
        DbType:   dbType,
        Host:     dbHost[0],
        Port:     dbPort,
        Database: models.DbCfg.Name,
        Username: models.DbCfg.User,
        Password: models.DbCfg.Passwd,
    }
    logger.Info("liquibase start with data source: %v", dataSource)

    // TODO why
    time.Sleep(10 * time.Second)

    migrate.Do(*migrateServerHost, "", "fcs", dataSource)

}

func HeaderProcess(h http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        tracing.SetHeader(r)
        resp := auditlog.NewRespProxyWriter(w)
        h.ServeHTTP(resp, r)
        auditlog.RecordOps(resp, r)
        tracing.RemoveHeader()
    })
}

func enrichSwaggerObject(swo *spec.Swagger) {
    swo.Info = &spec.Info{
        InfoProps: spec.InfoProps{
            Title:       "fcs",
            Description: "fcs-api",
        },
    }
}

func updateAdminToken() {
    var token *cksresource.UserAuthToken
    var err error
    if *cloudType == "public" {
        userInfo := &cksresource.UserPassword{
            UserName: model.Base.KnativeClusterUser,
            Password: model.Base.KnativeClusterPwd,
        }
        token, err = cksClient.Client.GetAuthToken(userInfo)
    } else {
        userInfo := &cksresource.UserName{
            User: model.Base.KnativeClusterUser,
        }
        token, err = cksClient.Client.GetTokenByClient(userInfo)
    }
    if err != nil {
        _ = logger.Error("GetAuthToken error: %v", err)
        return
    }
    model.Base.AdminToken = token.AccessToken
}

func prepareLcm(k8sIp string, api *handler.APIHandler) {
    fcClient, err := handler.GetFcClient(api)
    if err != nil {
        _ = logger.Error("prepareLcm: fcClient error: %v", err)
    }
    secrets, err := fcClient.GetSecrets("kube-system")
    if err != nil {
        _ = logger.Error("prepareLcm: fcClient error: %v", err)
    }
    var token string
    // 根据secret名字找到admin-token
    for _, secret := range secrets.Items {
        if strings.HasPrefix(secret.Name, "admin-token") {
            if tokenByte, ok := secret.Data["token"]; ok {
                token = base64.StdEncoding.EncodeToString(tokenByte)
            }
        }
    }
    // 重试3次防止意外
    err = backoff.RetryNotify(
        func() error {
            return resource.RegisterCluster(k8sIp, token)
        },
        backoff.WithMaxRetries(backoff.NewExponentialBackOff(), 3),
        func(err error, duration time.Duration) {
            _ = logger.Warn("failed register lcm err:%s,and it will be executed again in %v", err.Error(), duration)
        })
    if err != nil {
        _ = logger.Error("prepareLcm: register lcm err: %v", err)
    }
}

func initMinio() {
    minioClient, err := minio.New(*ossAdminHost, &minio.Options{
        Creds:  credentials.NewStaticV4(*minioAk, *minioSk, ""),
        Secure: false,
    })
    if err != nil {
        fmt.Println("minio.New err: ", err)
    }
    fmt.Printf("%#v\n", minioClient) // minioClient is now set up
    model.Base.MinioClient = minioClient
}

// 添加用于代理租户侧ip的内部域名
func initKongProxyService(ips []string) {
    ns := "fcs"
    config, err := clientcmd.BuildConfigFromFlags("", "")
    if err != nil {
        logger.Error("admin k8s config error: " + err.Error())
        return
    }
    clientSet, err := kube.NewForConfig(config)
    if err != nil {
        logger.Error("init admin k8s client set error: " + err.Error())
        return
    }
    // 手动创建svc和ep，其中ep的地址是租户侧集群三个管理网的地址。
    svc := &v1.Service{
        ObjectMeta: metav1.ObjectMeta{
            Name: "kong-proxy",
        },
        Spec: v1.ServiceSpec{
            Ports: []v1.ServicePort{
                {
                    Name:     "http",
                    Port:     80,
                    Protocol: v1.ProtocolTCP,
                    TargetPort: intstr.IntOrString{
                        Type:   intstr.Int,
                        IntVal: 80,
                    },
                },
            },
        },
    }
    _, err = clientSet.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
    if err != nil && !k8serrors.IsAlreadyExists(err) {
        logger.Error("create proxy svc error: " + err.Error())
    }

    ep := &v1.Endpoints{
        ObjectMeta: metav1.ObjectMeta{
            Name: "kong-proxy",
        },
        Subsets: []v1.EndpointSubset{
            {
                Addresses: []v1.EndpointAddress{},
                Ports: []v1.EndpointPort{
                    {
                        Name:     "http",
                        Port:     80,
                        Protocol: v1.ProtocolTCP,
                    },
                },
            },
        },
    }
    for _, ip := range ips {
        address := v1.EndpointAddress{IP: ip}
        ep.Subsets[0].Addresses = append(ep.Subsets[0].Addresses, address)
    }
    // 防止集群地址发送变化，所以每次重启前先删除ep
    _ = clientSet.CoreV1().Endpoints(ns).Delete(context.TODO(), "kong-proxy", metav1.DeleteOptions{})
    _, err = clientSet.CoreV1().Endpoints(ns).Create(context.TODO(), ep, metav1.CreateOptions{})
    if err != nil {
        logger.Error("create proxy endpoint error: " + err.Error())
    }
}

func startListenKsvc(api *handler.APIHandler, queue chan model.ListenInfo, stopCh <-chan struct{}) {
    // 持续更新informer
    go wait.Until(func() {
        newStopChan := make(chan struct{})
        cfg, err := handler.GetClientConfig(api)
        if err != nil {
            logger.Error("startListenKsvc get cfg error")
            return
        }
        ksvc.CreateInformer(cfg, newStopChan)
        // 结束当前 informer
        time.AfterFunc(time.Minute*11, func() {
            close(newStopChan)
        })
    }, time.Minute*10, stopCh)
    go ksvc.Run(queue)
}

func initI18n(defaultLanguage string) {
    path, _ := os.Executable()
    dir := filepath.Join(filepath.Dir(path), "i18n")
    _, err := ioutil.ReadDir(dir)
    if err != nil { //判断是否是go run调试
        _, path, _, _ := runtime.Caller(0)
        dir = filepath.Join(filepath.Dir(path), "i18n")
    }
    i18n.InitI18n(defaultLanguage, dir)
}
